package com.shujia.flink.tf

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

object Demo1Map {
  def main(args: Array[String]): Unit = {
    //创建flink环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")


    //java api
    val kvDS: DataStream[(String, Int)] = studentDS.map(new MapFunction[String, (String, Int)] {
      /**
       * 马匹方法，每一条数据执行一次，传进来一条返回一条
       *
       * @param value ： 一行数据
       * @return
       */
      override def map(value: String): (String, Int) = {
        val clazz: String = value.split(",")(4)
        (clazz, 1)
      }
    })

    kvDS.print()

    env.execute()

  }

}
